# Copyright (C) 2025 Kumo inc.
# Author: Jeff.li lijippy@163.com
# All rights reserved.
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published
# by the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program.  If not, see <https:#www.gnu.org/licenses/>.
#
"""Tests for test filtering protocol."""

import subprocess
import sys

from turbo import logging
from turbo.testing import _bazelize_command
from turbo.testing import turbotest
from turbo.testing import parameterized
from turbo.testing.tests import turbotest_env


@parameterized.named_parameters(
    ('as_env_variable_use_app_run', True, True),
    ('as_env_variable_no_argv', True, False),
    ('as_commandline_args_use_app_run', False, True),
    ('as_commandline_args_no_argv', False, False),
)
class TestFilteringTest(turbotest.TestCase):
  """Integration tests: Runs a test binary with filtering.

  This is done by either setting the filtering environment variable, or passing
  the filters as command line arguments.
  """

  def setUp(self):
    super().setUp()
    self._test_name = 'turbo/testing/tests/turbotest_filtering_test_helper'

  def _run_filtered(self, test_filter, use_env_variable, use_app_run):
    """Runs the py_test binary in a subprocess.

    Args:
      test_filter: string, the filter argument to use.
      use_env_variable: bool, pass the test filter as environment variable if
        True, otherwise pass as command line arguments.
      use_app_run: bool, whether the test helper should call
        `turbotest.main(argv=)` inside `app.run`.

    Returns:
      (stdout, exit_code) tuple of (string, int).
    """
    env = turbotest_env.inherited_env()
    env['USE_APP_RUN'] = '1' if use_app_run else '0'
    additional_args = []
    if test_filter is not None:
      if use_env_variable:
        env['TESTBRIDGE_TEST_ONLY'] = test_filter
      elif test_filter:
        # The -k flags are passed as positional arguments to turbo.flags.
        additional_args.append('--')
        additional_args.extend(['-k=' + f for f in test_filter.split(' ')])

    proc = subprocess.Popen(
        args=([_bazelize_command.get_executable_path(self._test_name)] +
              additional_args),
        env=env,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        universal_newlines=True)
    stdout = proc.communicate()[0]

    logging.info('output: %s', stdout)
    return stdout, proc.wait()

  def test_no_filter(self, use_env_variable, use_app_run):
    out, exit_code = self._run_filtered(None, use_env_variable, use_app_run)
    self.assertEqual(1, exit_code)
    self.assertIn('class B test E', out)

  def test_empty_filter(self, use_env_variable, use_app_run):
    out, exit_code = self._run_filtered('', use_env_variable, use_app_run)
    self.assertEqual(1, exit_code)
    self.assertIn('class B test E', out)

  def test_class_filter(self, use_env_variable, use_app_run):
    out, exit_code = self._run_filtered('ClassA', use_env_variable, use_app_run)
    self.assertEqual(0, exit_code)
    self.assertNotIn('class B', out)

  def test_method_filter(self, use_env_variable, use_app_run):
    out, exit_code = self._run_filtered('ClassB.testA', use_env_variable,
                                        use_app_run)
    self.assertEqual(0, exit_code)
    self.assertNotIn('class A', out)
    self.assertNotIn('class B test B', out)

    out, exit_code = self._run_filtered('ClassB.testE', use_env_variable,
                                        use_app_run)
    self.assertEqual(1, exit_code)
    self.assertNotIn('class A', out)

  def test_multiple_class_and_method_filter(self, use_env_variable,
                                            use_app_run):
    out, exit_code = self._run_filtered(
        'ClassA.testA ClassA.testB ClassB.testC', use_env_variable, use_app_run)
    self.assertEqual(0, exit_code)
    self.assertIn('class A test A', out)
    self.assertIn('class A test B', out)
    self.assertNotIn('class A test C', out)
    self.assertIn('class B test C', out)
    self.assertNotIn('class B test A', out)

  def test_substring(self, use_env_variable, use_app_run):
    out, exit_code = self._run_filtered(
        'testA', use_env_variable, use_app_run)
    self.assertEqual(0, exit_code)
    self.assertIn('Ran 2 tests', out)
    self.assertIn('ClassA.testA', out)
    self.assertIn('ClassB.testA', out)

  def test_glob_pattern(self, use_env_variable, use_app_run):
    out, exit_code = self._run_filtered(
        '__main__.Class*.testA', use_env_variable, use_app_run)
    self.assertEqual(0, exit_code)
    self.assertIn('Ran 2 tests', out)
    self.assertIn('ClassA.testA', out)
    self.assertIn('ClassB.testA', out)

  def test_not_found_filters_py37(self, use_env_variable, use_app_run):
    out, exit_code = self._run_filtered('NotExistedClass.not_existed_method',
                                        use_env_variable, use_app_run)
    if not use_env_variable and sys.version_info[:2] >= (3, 12):
      # When test filter is requested with the unittest `-k` flag, turbotest
      # respect unittest to fail when no tests run on Python 3.12+.
      self.assertEqual(5, exit_code)
    else:
      self.assertEqual(0, exit_code)
    self.assertIn('Ran 0 tests', out)

  def test_parameterized_unnamed(self, use_env_variable, use_app_run):
    out, exit_code = self._run_filtered('ParameterizedTest.test_unnamed',
                                        use_env_variable, use_app_run)
    self.assertEqual(0, exit_code)
    self.assertIn('Ran 2 tests', out)
    self.assertIn('parameterized unnamed 1', out)
    self.assertIn('parameterized unnamed 2', out)

  def test_parameterized_named(self, use_env_variable, use_app_run):
    out, exit_code = self._run_filtered('ParameterizedTest.test_named',
                                        use_env_variable, use_app_run)
    self.assertEqual(0, exit_code)
    self.assertIn('Ran 2 tests', out)
    self.assertIn('parameterized named 1', out)
    self.assertIn('parameterized named 2', out)


if __name__ == '__main__':
  turbotest.main()
